package Source;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import utils.MySqlUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * Create by zhh on 2022/9/6.
 */
public class MySqlSource extends RichSourceFunction<Student> {

    Connection connection;
    PreparedStatement psmt;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection= MySqlUtils.getConnection();
        psmt = connection.prepareStatement("select * from scores");
    }

    @Override
    public void close() throws Exception {
        MySqlUtils.close(connection,psmt);
    }

    @Override
    public void run(SourceContext<Student> ctx) throws Exception {
        ResultSet rs = psmt.executeQuery();
        while (rs.next()){
            int id = rs.getInt("id");
            String name = rs.getString("name");
            String course = rs.getString("course");
            int scores = rs.getInt("scores");
            ctx.collect(new Student(id,name,course,scores));
        }
    }

    @Override
    public void cancel() {

    }
}
